package com.bbx.flink.demo.window;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.ProcessingTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.util.Map;

public class SessionWindowDemo {
    public static void main(String [] args) throws Exception {
        //获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //获取输入流
        env.socketTextStream("192.168.10.131",10003)
                .map((MapFunction<String, Integer>) value -> {
                    return  Integer.parseInt(value);
                }).keyBy(i->i%2)
                .window(ProcessingTimeSessionWindows.withGap(Time.seconds(20)))
                .sum(0)
                .print();

        env.execute();
    }
}
